package com.dingdang.mq;

import com.aliyun.dysmsapi20170525.Client;
import com.aliyun.dysmsapi20170525.models.SendSmsRequest;
import com.aliyun.teaopenapi.models.Config;
import com.dingdang.service.SendMessagesServiceI;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "smsTopic",consumerGroup = "smsGroup")
public class SendsPhoneMessagesMQ implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private SendMessagesServiceI sendMessagesServiceI;
    @Override
    public void onMessage(String s) {
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {

//                  解决幂等性问题
                    Boolean aBoolean = redisTemplate.boundHashOps("smsMessage").hasKey(msg.getMsgId());

                    try {
                        if (!aBoolean){
                                String phone = new String(msg.getBody());
//                                发短信
                                sendMessagesServiceI.sendMessage(phone);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        /*死信队列*/
                        System.out.println("出现异常" + e.getMessage());
                        if (msg.getReconsumeTimes() > 2) {
                            consumeConcurrentlyContext.setDelayLevelWhenNextConsume(-1);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
    public Client createClient (String accessKeyId, String accessKeySecret) throws Exception {
        Config config = new Config()
                // 您的AccessKey ID
                .setAccessKeyId(accessKeyId)
                // 您的AccessKey Secret
                .setAccessKeySecret(accessKeySecret);
        // 访问的域名
        config.endpoint = "dysmsapi.aliyuncs.com";
        return new Client(config);
    }
}
